package com.example.springboottest.example.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.AsyncCallback;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.function.Consumer;

/**
 * 因为 localhost:2181 localhost:2182 localhost:2183
 * 是集群搭建的，所以随便操作哪个节点是一样的效果
 */
public class ZKCuratorTest {
    public static void main(String[] args) {
        String zkPath = "/test/CRUD/node-2";
        String data = "Hello World!";
//        createNode(zkPath);

//        readNode(zkPath);

//        updateNode(zkPath, data);
//        updateNodeAsync(zkPath, data);

        deleteNode(zkPath);
    }

    public static void readNode(String zkPath) {
        String zkAddress = "localhost:2182";
        Consumer<CuratorFramework> consumer = client -> {
            try {
                Stat stat = client.checkExists().forPath(zkPath);
                if (null != stat) {
                    //读取节点的数据
                    byte[] payload = client.getData().forPath(zkPath);
                    String data = new String(payload, "UTF-8");
                    System.out.println("read data:" + data);
                    String parentPath = "/test";
                    List<String> children = client.getChildren().forPath(parentPath);
                    for (String child : children)
                        System.out.println("child:" + child);
                }
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        handleNode(zkAddress, consumer);
    }

    public static void updateNode(String zkPath, String data) {
        String zkAddress = "localhost:2183";
        handleNode(zkAddress, client -> {
            try {
                //同步更新
                client.setData().forPath(zkPath, data.getBytes("UTF-8"));
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    /*
    如果是异步更新呢？
    其实很简单，通过 SetDataBuilder 构造者实例的 inBackground(AsyncCallback callback)方法，
    设置一个 AsyncCallback 回调实例。简简单单的一个函数，就将更新数据的行为从同步执行变成了异步执行。
     */
    public static void updateNodeAsync(String zkPath, String data) {
        String zkAddress = "localhost:2183";
        handleNode(zkAddress, client -> {
            try {
                AsyncCallback.StringCallback callback
                        = new AsyncCallback.StringCallback() {
                    //回调方法
                    @Override
                    public void processResult(int i, String s, Object o, String s1) {
                        System.out.println(
                                "i = " + i + " | " +
                                        "s = " + s + " | " +
                                        "o = " + o + " | " +
                                        "s1 = " + s1
                        );
                    }
                };

                client.setData().
                        inBackground(callback)//设置回调实例
                        .forPath(zkPath, data.getBytes("UTF-8"));

                Thread.sleep(10000);  //模拟休息10s,不然异步修改不会成功
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    public static void deleteNode(String zkPath) {
        String zkAddress = "localhost:2183";
        handleNode(zkAddress, client -> {
            try {
                client.delete().forPath(zkPath);
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }

    public static void createNode(String zkPath) {
        String zkAddress = "localhost:2181";
        Consumer<CuratorFramework> consumer = client -> {
            try {
                String data = "hello";
                byte[] payload = new byte[0];
                payload = data.getBytes("UTF-8");
                client.create()
                        .creatingParentsIfNeeded()
                        .withMode(CreateMode.PERSISTENT)
                        .forPath(zkPath, payload);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            }
        };

        handleNode(zkAddress, consumer);
    }

    public static void handleNode(String zkAddress, Consumer<CuratorFramework> consumer) {
        CuratorFramework client = ClientFactory.createSimple(zkAddress);
        try {
            client.start();

            consumer.accept(client);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            CloseableUtils.closeQuietly(client);
        }
    }
}
